package com.test.rxjavademo;

import org.jetbrains.annotations.NotNull;

import io.reactivex.Observable;
import io.reactivex.ObservableEmitter;
import io.reactivex.ObservableOnSubscribe;
import io.reactivex.Observer;
import io.reactivex.Scheduler;
import io.reactivex.android.schedulers.AndroidSchedulers;
import io.reactivex.disposables.Disposable;
import io.reactivex.functions.Consumer;
import io.reactivex.functions.Function;
import io.reactivex.schedulers.Schedulers;

class ToolOperatorDemo {

    private static Observer observer = new Observer<Object>() {//Observer观察者，订阅subscribe了被观察者Observable

        @Override
        public void onSubscribe(@NotNull Disposable disposable) {
            System.out.println("1、onSubscribe,首次订阅时会回调:" + disposable);
        }

        @Override
        public void onNext(@NotNull Object o) {
            System.out.println("onNext:" + o);
        }

        @Override
        public void onError(@NotNull Throwable throwable) {
            System.out.println("onError:" + throwable);
        }

        @Override
        public void onComplete() {
            System.out.println("onComplete");
        }
    };






    public static void test(){
        //
        Observable.create(new ObservableOnSubscribe<Object>() {
                              @Override
                              public void subscribe(@NotNull ObservableEmitter<Object> e) throws Exception {
//                                  Thread.sleep(2000);
                                  System.out.println("subscribe:" + Thread.currentThread().getName());//RxNewThreadScheduler-1
                                  e.onNext(1);
                                  e.onNext(2);
                                  e.onComplete();
                              }
                          }

        )           .subscribeOn(Schedulers.newThread())//来决定执行subscribe所在线程，即事件产生所有现在
                .observeOn(AndroidSchedulers.mainThread())
                .doOnNext(new Consumer<Object>() {
            @Override
            public void accept(@NotNull Object o) throws Exception {
                System.out.println("doOnNext accept:"+ o + " "+ Thread.currentThread().getName());//RxNewThreadScheduler-1
            }
        })
//                .subscribeOn(Schedulers.newThread())//来决定执行subscribe所在线程，即事件产生所有现在
//                .observeOn(AndroidSchedulers.mainThread())
                .map(new Function<Object, Object>() {
                    @NotNull
                    @Override
                    public Object apply(@NotNull Object o) throws Exception {
                        System.out.println("map apply:" + Thread.currentThread().getName());//RxNewThreadScheduler-1
                        return o + "1";
                    }
                })
//                .observeOn(Schedulers.io())
                .subscribe(new Observer<Object>() {
                    @Override
                    public void onSubscribe(@NotNull Disposable disposable) {
                        System.out.println("onSubscribe:" + Thread.currentThread().getName());//main
                    }

                    @Override
                    public void onNext(@NotNull Object o) {
                        System.out.println("onNext:" + o + " " +  Thread.currentThread().getName());//RxNewThreadScheduler-1
                    }

                    @Override
                    public void onError(@NotNull Throwable throwable) {
                        System.out.println("onError:" + Thread.currentThread().getName());
                    }

                    @Override
                    public void onComplete() {
                        System.out.println("onComplete:" + Thread.currentThread().getName());
                    }
                });

        try {
            Thread.sleep(2000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        }
    }




    public static void main(String[] args) {
        test();
    }

}
